package com.wuwangfu.join;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @Author jcshen
 * @Date 2023-02-27
 * @PackageName:com.wuwangfu.join
 * @ClassName: TumbleWindowLeftOuterJoin
 * @Description: 使用左外连接，第一个流如果没有join上，也要输出
 * @Version 1.0.0
 */
public class TumbleWindowLeftOuterJoin {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        /**
         * 1000,A,aaa
         * 2000,B,aaa
         * 4999,C,aaa
         */
        DataStreamSource<String> left = env.socketTextStream("localhost", 8888);
        /**
         * 1200,A,111
         * 3333,B,222
         * 4999,A,333
         */
        DataStreamSource<String> right = env.socketTextStream("localhost", 9999);

        //生成watermark
        SingleOutputStreamOperator<String> leftWithWatermark = left.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });

        SingleOutputStreamOperator<String> rightWithWatermark = right.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        //map
        SingleOutputStreamOperator<Tuple3<Long, String, String>> leftmaped = leftWithWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });

        SingleOutputStreamOperator<Tuple3<Long, String, String>> rightmaped = rightWithWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });
        //第一个流（左流）和第二个流（右流）进行leftOuterjoin
        //在同一个窗口内，并且join条件相等，第一个流的数据没join上也输出
        leftmaped.coGroup(rightmaped)
                .where(t -> t.f1)
                .equalTo(t -> t.f1)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple3<Long, String, String>, Tuple3<Long, String, String>, Tuple5<Long, String, String, Long, String>>() {
                    @Override
                    public void coGroup(Iterable<Tuple3<Long, String, String>> first, Iterable<Tuple3<Long, String, String>> second, Collector<Tuple5<Long, String, String, Long, String>> out) throws Exception {
                        //当两个流窗口满足触发条件，就会调用coGroup方法，输入的数据是同一个分区中的数据
                        //如果第一个迭代器中有数据，第二个迭代器也有数据join上了
                        boolean flag = false;
                        for (Tuple3<Long, String, String> tf : first) {
                            for (Tuple3<Long, String, String> ts : second) {
                                flag = true;
                                //数据输出
                                out.collect(Tuple5.of(tf.f0, tf.f1, tf.f2, ts.f0, ts.f2));
                            }
                            //第一个迭代器中有数据，第二个迭代器中没有数据，leftOuterJoin
                            if (!flag) {
                                out.collect(Tuple5.of(tf.f0, tf.f1, tf.f2, null, null));
                            }
                        }


                    }
                }).print();

        env.execute();
    }
}
